package com.producer;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import javax.annotation.PostConstruct;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;

import com.util.JsonMapper;

@Service
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
@ConfigurationProperties(prefix="kafka")
public class KafkaProducerService {
	 private  KafkaProducer<String, String> producer;
	 private  String servers;
	 
	 public String getServers() {
		return servers;
	}

	public void setServers(String servers) {
		this.servers = servers;
	}
	@PostConstruct
	void init(){
		    Properties props = new Properties();
	        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
	        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "piwik_to_pio_producer");
	        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	        props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
	        producer = new KafkaProducer<>(props);
	      
	 }
	JsonMapper jsonMapper =JsonMapper.nonDefaultMapper();
	 public void sendMessgae(String topic,String key,Map<String, Object> data) throws InterruptedException, ExecutionException{		
		 
		 try {
			sendMessgae(topic, key, jsonMapper.toJson(data));
		} catch (InterruptedException | ExecutionException e) {
			try {
				sendMessgae(topic, key, jsonMapper.toJson(data));
			} catch (InterruptedException | ExecutionException e1) {
					sendMessgae(topic, key, jsonMapper.toJson(data));
				e1.printStackTrace();
			}
			e.printStackTrace();
		}
	 }
	 
	 public void sendMessgae(String topic,String key,String data) throws InterruptedException, ExecutionException{		 
		 producer.send(new ProducerRecord<>(topic, key,data)).get();
	 }
}
